package com.rrd.hive.udf;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.ql.exec.UDF;
import org.apache.hadoop.io.Text;

import java.net.URI;
import java.util.*;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStreamReader;

public class MappingUDF extends UDF  {
    Text text = new Text();

    String sys_mapping_code = null;
    HashMap<String, String> code_maping = new HashMap<String, String>();
    public Text evaluate(String code_type,String table_name,String src_code) throws Exception{

       if(sys_mapping_code == null) {
           Configuration conf = new Configuration();
           String code_type_get = "/data/cdm/" + code_type + "/000000_0";
           String filePath = "/data/cdm/t07_minor_sys_source_pa_all/000000_0";
           String sys_code = StringUtils.substringBefore(table_name, "_");
           int fiel_type = 1;
           HashMap<String, String> maping = readTxtFile(filePath,conf,fiel_type);
           sys_mapping_code = maping.get(sys_code);
           if(sys_mapping_code == null){
               throw new Exception("未获得源系统映射值");
           }
           List<String> type_one = new ArrayList<>(Arrays.asList("t07_funder_mapping_pa_all", "t07_mgmt_fee_type_mapping_pa_all", "t07_repay_mode_mapping_pa_all"
                   , "t07_gender_mapping_pa_all", "t07_education_mapping_pa_all"));
           List<String> type_two = new ArrayList<>(Arrays.asList("t07_industry_mapping_pa_all"));
           List<String> type_three = new ArrayList<>(Arrays.asList("t02_loan_product_mapping_d_all"));
           List<String> type_four = new ArrayList<>(Arrays.asList("t05_region_mapping_pa_all"));
           int fiel_type2;
           if (type_one.contains(code_type)) {
               fiel_type2 = 2;
           } else if (type_two.contains(code_type)) {
               fiel_type2 = 3;
           }else if (type_three.contains(code_type)) {
               fiel_type2 = 4;
           }else if (type_four.contains(code_type)) {
               fiel_type2 = 5;
           }  else {
               fiel_type2 = 6;
           }
           code_maping = readTxtFile(code_type_get,conf, fiel_type2);
       }
        String src_key = sys_mapping_code+src_code;
        String result = code_maping.get(src_key.trim());
        //String result = sys_mapping_code;
        if(result == null){
            result = "99";
        }
        text.set(result);
        return text;
    }

    public static HashMap<String, String> readTxtFile(String filePath,Configuration conf,int fiel_type) throws Exception{
        HashMap<String, String> map =  new HashMap();
        try {
            FileSystem fs = FileSystem
                    .newInstance(URI.create(filePath), conf);
            Path remotePath = new Path(filePath);
            FSDataInputStream in = fs.open(remotePath);
            if (fs.exists(remotePath)) {
                BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(in));
                String lineTxt = null;
                //记录读取的数据文件的行数
                int count = 0;
                //定义字段的数组
                String[] fields = null;
                //定义每条记录取出的字段值数组
                String[] fieldValue = null;
                //定义Map集合
                if (fiel_type == 1) {
                    while ((lineTxt = bufferedReader.readLine()) != null) {
                        fieldValue = lineTxt.split(",");
                        map.put(fieldValue[8], fieldValue[0]);
                        count++;
                    }
                } else if (fiel_type == 2) {
                    while ((lineTxt = bufferedReader.readLine()) != null) {
                        fieldValue = lineTxt.split(",");
                        String key_code = fieldValue[3] + fieldValue[4];
                        map.put(key_code, fieldValue[0]);
                        count++;
                    }
                } else if (fiel_type == 3) {
                    while ((lineTxt = bufferedReader.readLine()) != null) {
                        fieldValue = lineTxt.split(",");
                        String key_code = fieldValue[2] + fieldValue[3];
                        map.put(key_code, fieldValue[0]);
                        count++;
                    }
                }else if (fiel_type == 4) {
                    while ((lineTxt = bufferedReader.readLine()) != null) {
                        fieldValue = lineTxt.split("\\t");
                        String key_code = fieldValue[3] + fieldValue[4];
                        map.put(key_code, fieldValue[0]);
                        count++;
                    }
                }else if (fiel_type == 5) {
                    while ((lineTxt = bufferedReader.readLine()) != null) {
                        fieldValue = lineTxt.split("\\t");
                        String key_code = fieldValue[2] + fieldValue[3];
                        map.put(key_code, fieldValue[0]);
                        count++;
                    }
                } else {
                    System.out.println("不存在的代码类型");
                }
                bufferedReader.close();
                in.close();
                fs.close();
            } else {
                System.out.println("找不到指定的文件");
                throw new Exception("找不到指定的文件");
            }
        }catch (Exception e) {
        System.out.println("读取文件内容出错");
        e.printStackTrace();
        throw new Exception("读取文件内容出错");
    }
        return map;
    }
    /*public static HashMap<String, String> readTxtFile2(String filePath,int fiel_type){
        //定义存储读取到的数据记录的集合
        HashMap<String, String> map =  new HashMap();
        try {
            String encoding="UTF-8";
            File file=new File(filePath);
            //判断文件是否存在
            if(file.isFile() && file.exists()){
                //考虑到编码格式
                InputStreamReader read = new InputStreamReader(new FileInputStream(file),encoding);
                BufferedReader bufferedReader = new BufferedReader(read);
                String lineTxt = null;
                //记录读取的数据文件的行数
                int count = 0;
                //定义字段的数组
                String[] fields = null;
                //定义每条记录取出的字段值数组
                String[] fieldValue = null;
                //定义Map集合
                if(fiel_type==1) {
                    while ((lineTxt = bufferedReader.readLine()) != null) {
                        fieldValue = lineTxt.split(",");
                        //System.out.println("fieldValue:"+fieldValue);
                        map.put(fieldValue[8], fieldValue[0]);
                        //将读取的每一行的记录存入到list集合中
                        count++;
                    }
                }else if(fiel_type==2) {
                    while ((lineTxt = bufferedReader.readLine()) != null) {
                        fieldValue = lineTxt.split(",");
                        //System.out.println("fieldValue:"+fieldValue);
                        String key_code = fieldValue[3]+fieldValue[4];
                        map.put(key_code, fieldValue[0]);
                        //将读取的每一行的记录存入到list集合中
                        count++;
                    }
                }else if(fiel_type==3) {
                    while ((lineTxt = bufferedReader.readLine()) != null) {
                        fieldValue = lineTxt.split(",");
                        //System.out.println("fieldValue:"+fieldValue);
                        String key_code = fieldValue[2]+fieldValue[3];
                        map.put(key_code, fieldValue[0]);
                        //将读取的每一行的记录存入到list集合中
                        count++;
                    }
                }else{
                    System.out.println("不存在的代码类型");
                }
                read.close();
            }else{
                System.out.println("找不到指定的文件");
            }
        } catch (Exception e) {
            System.out.println("读取文件内容出错");
            e.printStackTrace();
        }
        return map;
    }*/

    /*public static void main(String[] args) throws Exception {
        String args1 = "t07_gender_mapping_pa_all";
        String args2 = "thd01_lend_customers_chain";
        String args3 = "FEMALE";
        MappingUDF dfu = new MappingUDF();
        System.out.println(dfu.evaluate(args1,args2,args3));
    }*/


}